#include "gateway_stream.h"
#include "gateway_network_event.h"
#include <boost/lexical_cast.hpp>
#include <boost/interprocess/detail/atomic.hpp>

using namespace boost;
using namespace boost::asio;
using namespace boost::asio::ip;

const int default_buffer_size = 1024;// 默认1024

gateway_stream::gateway_stream(io_service& ioservice)
: _cache(0)
, _total(default_buffer_size)
, _timeout(10)
, _buffer(NULL)
, _error_state(0)
, _io_strand(ioservice)
, _stream_event(NULL)
, _timer(ioservice)
, _socket(ioservice)
, _last_readwrite_time(posix_time::second_clock::local_time()){

}

gateway_stream::~gateway_stream(){
	if (_buffer) delete []_buffer;
}

tcp::socket& gateway_stream::get_socket(){
	return _socket;
}

void gateway_stream::open(){	
	if (_stream_event) _stream_event->stream_init();

	static tcp::no_delay option(true);
	_socket.set_option(option);

	_buffer = new unsigned char[_total];
	_socket.async_read_some(buffer(_buffer, _total), _io_strand.wrap(bind(&gateway_stream::handle_read, shared_from_this(),
	boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)));

	_last_readwrite_time = posix_time::second_clock::local_time();
	_timer.expires_from_now(posix_time::seconds(_timeout));
	_timer.async_wait(_io_strand.wrap(bind(&gateway_stream::handle_timer, shared_from_this(), boost::asio::placeholders::error)));
}

void gateway_stream::close(){
	_io_strand.post(boost::bind(&gateway_stream::handle_idle_connection, shared_from_this(), boost::asio::error::shut_down));
}

bool gateway_stream::is_open() {
	return _socket.is_open();
}

void gateway_stream::do_error(bool inner){
	if (interprocess::ipcdetail::atomic_cas32(&_error_state, 1, 0) == 0) {
		system::error_code ec;
		_socket.shutdown(tcp::socket::shutdown_both, ec);
		_socket.close(ec);
		_timer.cancel(ec);
	}

	if (inner && _stream_event){
		_stream_event->stream_clean();
	}
}

void gateway_stream::set_event_handle(gateway_stream_event* stream_event){
	_stream_event = stream_event;
}

void gateway_stream::handle_read(const system::error_code& error, size_t transferred){
	if (error) {
		do_error(true);
		return;
	}

	_last_readwrite_time = posix_time::second_clock::local_time();
	if (_stream_event) {
		unsigned int used = _stream_event->stream_read(_buffer, _cache + transferred);
		if (used > 0){
			used = used <= _cache + transferred ? used : _cache + transferred;
			if (_cache + transferred - used > 0) {
				memmove(_buffer, _buffer + used, _cache + transferred - used);
			}
				
			_cache = _cache + transferred - used;
		}	
		else {
			_cache += transferred;
		}
	}

	if (5 * _cache >= 4 * _total) {// 超过80%开始扩容
		assign_copy();
	}
	else if (_cache == 0 && _total != default_buffer_size) {
		delete []_buffer;
		_total = default_buffer_size;
		_buffer = new unsigned char[_total];
	}

	_socket.async_read_some(buffer(_buffer + _cache, _total - _cache), _io_strand.wrap(bind(&gateway_stream::handle_read, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)));
}

void gateway_stream::assign_copy(){//呈2倍数扩容
	unsigned char* buffer = new unsigned char[_total*2];
	memcpy(buffer, _buffer, _cache);
	delete []_buffer;
	_buffer = buffer;
	_total *= 2;
}

int gateway_stream::send(unsigned char* data, unsigned int len){
	
	int sent = 0;
	int total = len;
	system::error_code ec;
	while (_socket.is_open() && len > 0) {
		sent = _socket.send(buffer(data, len), 0, ec);
		if (sent > 0) {
			data += sent;
			len -= sent;
		}
		else {
			break;
		}
	}

	return total - len;
}

void gateway_stream::set_timeout(unsigned int timeout){
	_timeout = timeout;
	_io_strand.post(boost::bind(&gateway_stream::handle_idle_connection, shared_from_this(), system::error_code()));
}

void gateway_stream::handle_timer(const system::error_code& error){
	if (error) return;// cancel
	_io_strand.post(boost::bind(&gateway_stream::handle_idle_connection, shared_from_this(), error));
}

void gateway_stream::handle_idle_connection(const system::error_code& error){
	if (error) { // shutdown
		do_error(false);
		return;
	}

	if (_timeout != 0) {
		posix_time::time_duration duration = posix_time::second_clock::local_time() - _last_readwrite_time;
		int new_timeout = _timeout - duration.total_seconds();
		if (new_timeout <= 0){ // 超时关闭连接
			do_error(false);
		}
		else {
			_timer.expires_from_now(posix_time::seconds(new_timeout));
			_timer.async_wait(_io_strand.wrap(bind(&gateway_stream::handle_timer, shared_from_this(), boost::asio::placeholders::error)));
		}
	}
}

std::string gateway_stream::get_remote_host(){
	boost::system::error_code ec;
	tcp::endpoint ep = _socket.remote_endpoint(ec);
	if (!ec){
		return ep.address().to_string();
	}
	else {
		return "";
	}
}

unsigned short gateway_stream::get_remote_port(){
	boost::system::error_code ec;
	tcp::endpoint ep = _socket.remote_endpoint(ec);
	if (!ec){
		return ep.port();
	}
	else {
		return 0;
	}
}

